package cn.zhaopin.starter.mq.support;

import com.fasterxml.jackson.databind.JavaType;
import org.springframework.core.GenericTypeResolver;
import org.springframework.core.MethodParameter;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.util.ClassUtils;

import java.io.IOException;
import java.lang.reflect.Type;

/**
 * description:jackson扩展
 *
 * @author: zuomin (myleszelic@outlook.com)
 * @date 2021/07/31:23:09
 */
public class PulsarJackson2MessageConverter extends MappingJackson2MessageConverter {

    @Override
    protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
        JavaType javaType = getObjectMapper().constructType(getResolvedType(targetClass, conversionHint));
        Object payload = message.getPayload();
        Class<?> view = getSerializationView(conversionHint);
        try {
            if (ClassUtils.isAssignableValue(targetClass, payload)) {
                return payload;
            }
            else {
                // Assuming a text-based source payload
                if (view != null) {
                    return getObjectMapper().readerWithView(view).forType(javaType).readValue(getObjectMapper().writeValueAsString(payload));
                }
                else {
                    return getObjectMapper().readValue(getObjectMapper().writeValueAsString(payload), javaType);
                }
            }
        }
        catch (IOException ex) {
            throw new MessageConversionException(message, "Could not read JSON: " + ex.getMessage(), ex);
        }
    }

    @Override
    protected Object convertToInternal(Object payload, MessageHeaders headers, Object conversionHint) {
        return super.convertToInternal(payload, headers, null);
    }

    static Type getResolvedType(Class<?> targetClass, @Nullable Object conversionHint) {
        if (conversionHint instanceof MethodParameter) {
            MethodParameter param = (MethodParameter) conversionHint;
            param = param.nestedIfOptional();
            if (Message.class.isAssignableFrom(param.getParameterType())) {
                param = param.nested();
            }
            Type genericParameterType = param.getNestedGenericParameterType();
            Class<?> contextClass = param.getContainingClass();
            return GenericTypeResolver.resolveType(genericParameterType, contextClass);
        }
        return targetClass;
    }
}
